package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo17Cache {
  def main(args: Array[String]): Unit = {
    // 创建spark环境
    val conf = new SparkConf()
    //local[*]： 使用计算所有的资源（CPU）
    conf.setMaster("local[*]")
    conf.setAppName("student")

    val sc = new SparkContext(conf)

    //指定checkpoint路径
    //sc.setCheckpointDir("spark/data/checkpoint")

    //读取数
    val linesD: RDD[String] = sc.textFile("spark/data/student_bigdata.csv")

    //解析数据
    val studentRDD: RDD[(String, String, Int, String, String)] = linesD
      .map(line => {
        //println("map被执行")
        val split: Array[String] = line.split(",")
        val id: String = split(0)
        val name: String = split(1)
        val age: Int = split(2).toInt
        val sex: String = split(3)
        val clazz: String = split(4)
        (id, name, age, sex, clazz)
      })


    //缓存RDD,默认缓存策略时MEMORY_ONLY
    //studentRDD.cache()


    /**
     * 缓存： 将数据缓存到Executor的内存或者磁盘上，任务结束数据就没了
     * persist: 可以手动指定缓存策略
     * 1、MEMORY_ONLY： 仅内存， 速度快，如果数据量大，内存放不下
     * 2、MEMORY_AND_DISK_SER: 内存放不下方磁盘，同时对数据进行压缩
     *
     * 为什么不使用MEMORY_AND_DISK， 因为不管压不压缩，内存都比磁盘快，所以尽量将数据压到内存中
     *
     *
     * 压缩：时间换空间（CPu的计算时间换取存储空间），解压和压缩需要CPU计算时间
     *
     */
    studentRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)


    /**
     * checkpoint: 将数据缓存到HDFS中，在后面学习Spark Streaming时会用到
     * 需要指定checkpoint路径
     */
    //studentRDD.checkpoint()


    //1、统计班级的人数
    studentRDD
      .map(stu => (stu._5, 1))
      .reduceByKey((x, y) => x + y)
      .foreach(println)

    //2、统计性别的人数
    studentRDD
      .map(stu => (stu._4, 1))
      .reduceByKey((x, y) => x + y)
      .foreach(println)

    //    while (true) {}

  }
}
